package com.atguigu.gmall.realtime.utils;

import com.atguigu.gmall.realtime.bean.TransientSink;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Description: 往ClickHouse中插入数据的工具类
 * @Author: tiancy
 * @Create: 2021/12/7
 */
public class ClickHouseUtil {

    /**
     * 给定一个sql,获取一个 SinkFunction示例,当前 sinkFunction会往ClickHouse中的一张表中插入数据.
     *
     * @param sql 给定的向某张表中插入的SQL
     * @param <T> 需要做的实体关系映射的具体`实体对象`.这里指的是流中`访客主题中的指标`
     * @return
     */
    public static <T> SinkFunction<T> getSinkFunction(String sql) {
        /*
            参数1： 传入Sql，格式如：insert into xxx values(?,?,?,?)
            参数2:  可以用lambda表达实现(jdbcPreparedStatement, t) -> t为数据对象，要装配到语句预编译器的参数中。
            参数3：设定一些执行参数，比如重试次数，批次大小。
            参数4：设定连接参数，比如地址，端口，驱动名。
         */
        return JdbcSink.sink(
                sql,
                // 如何给问号占位符赋值
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement ps, T obj) throws SQLException {
                        //给sql中的问号占位符赋值  insert into visitor_stats_0609 values(?,?,?,?,?,?,?,?,?,?,?,?)
                        /*
                            TODO 给 实体和SQL中的占位符,做对象关系映射(ORM).
                             但是在使用前有一个前提,需要我们的建表语句中字段值和我们实体类中的字段值顺序必须一致.
                         */
                        //获取流中对象的类型  --- 通过类型获取类中的属性,包括私有属性
                        Field[] fieldArr = obj.getClass().getDeclaredFields();
                        //对属性进行遍历
                        int skipNum = 0;
                        for (int i = 0; i < fieldArr.length; i++) {
                            //每遍历一次  获取一个属性对象
                            Field field = fieldArr[i];

                            //判断当前属性是否需要保存到CK中
                            TransientSink transientSink = field.getAnnotation(TransientSink.class);
                            if (transientSink != null) {
                                skipNum++;
                                //如果对象不为空 ，说明属性上有@TransientSink注解标注，不需要保存到CK中
                                continue;
                            }

                            //设置属性的访问权限
                            field.setAccessible(true);
                            try {
                                //获取流中对象对应的属性值   正常:对象.属性    反射: 属性对象.get(对象)
                                Object fieldValue = field.get(obj);
                                //给问号占位符赋值
                                ps.setObject(i + 1 - skipNum, fieldValue);
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                },
                // 设置批次数量大小,每个单独的并行度批次大小达到多少后,开始写入olap`clickhouse`中.
                JdbcExecutionOptions.builder()
                        .withBatchSize(5)
                        .build(),
                new JdbcConnectionOptions
                        .JdbcConnectionOptionsBuilder()
                        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                        .withUrl(GmallConfig.CLICKHOUSE_URL).build()
        );
    }
}
